package org.zjt.spark.book

import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayBuffer

/**
  *  找出所有三角形：
  *   (k,任意v,_) 确定可能形成三角形的边是否相连。
  *
  *     1、建立相互的边    a,b 得到 a,b和b,a
  *     2、找到每个节点的相连的点，作为三联的直接边的依据
  *     3、三联三角形   直接相连 (k,任意v,_)     可能相连的边三角形（任意v,任意V，k）
  *     4、reduce包含(k,任意v,_)的三角形才是边相连的。
  *     5、去除重复的。
  *
  */
object FindTriangle extends App {
  var sparkConf = new SparkConf().setMaster("local[2]").setAppName("WeatherSort")
  val sc = new SparkContext(sparkConf)

  val rdd = sc.textFile("D:\\Idea workspace\\scala-demo\\src\\main\\resource\\triangle.data").flatMap {
    line => {
      var data = ArrayBuffer[(String, String)]();
      if (!line.startsWith("#")) {
        val array = line.split("   ")
        data += new Tuple2(array.head, array.last)
        data += new Tuple2(array.last, array.head)
      }
      data
    }
  }.filter(_ != null).groupByKey().flatMap {
    line => {
      val data = new ArrayBuffer[(String, String)]()

      //可能相连的边，但可能不相连
      for (item1 <- line._2) {
        for (item2 <- line._2) {
          if (item1 != item2) {
            data += new Tuple2("%s:%s".format(item1, item2), line._1) //可能三角形
          }
        }
      }

      //直接相连的边
      for (item1 <- line._2) {
        data += new Tuple2("%s:%s".format(line._1, item1), "_")
      }
      data
    }
  }.groupByKey().flatMap{
    line => {
      val data = new ArrayBuffer[(String, String)]()
      if(line._2.toArray.contains("_")){
        for(item <- line._2 if !item.equals("_") ){
          var sort = new ArrayBuffer[Int]()
          sort += item.toInt
          sort ++= line._1.split(":").map(_.toInt)
          sort = sort.sorted
          data += new Tuple2("%s:%s".format(sort(0), sort(1)), sort(2)+"")
        }
      }
      data
    }
  }.distinct()


  rdd.foreach(println)

  sc.stop()

}
